package com.zk.kafka;

import com.csvreader.CsvReader;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.nio.charset.Charset;
import java.util.Properties;

public class CsvDataProducer {
    //kafka配置信息
    private static final String BOOTSTRAP_SERVERS = "localhost:9092";//端口
    private static final String TOPIC_NAME = "test";//主题

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); //串行化调度
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.ACKS_CONFIG, "1");  // 确保消息不丢失
        props.put(ProducerConfig.RETRIES_CONFIG, 3); //重试次数

        //CSV文件读取
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            CsvReader csvReader = new CsvReader("src/main/resources/data.csv", ',', Charset.forName("UTF-8"));
            csvReader.readHeaders(); //读取头部，即跳过该段

            while (csvReader.readRecord()) {
                String lastCounter = csvReader.get("last_counter");  // 日期
                String agent = csvReader.get("agent"); // 浏览器类型
                String platform = csvReader.get("platform"); // 操作系统类型
                String device = csvReader.get("device"); // 设备
                String ip = csvReader.get("ip"); // IP地址
                int hits = Integer.parseInt(csvReader.get("hits")); // 点击次数

                // 拼接为JSON格式
                String message = String.format("{\"lastCounter\":\"%s\",\"ip\":\"%s\",\"agent\":\"%s\"," +
                                "\"platform\":\"%s\",\"device\":\"%s\",\"hits\":%s}",
                        lastCounter, ip, agent, platform, device,hits);

                ProducerRecord<String, String> record = new ProducerRecord<>(
                        TOPIC_NAME,
                        lastCounter,  // 以lastCounter为key，确保相同以lastCounter为key的数据到同一分区
                        message
                );
                
                //发送消息
                producer.send(record, (metadata, exception) -> {
                    if (exception != null) {
                        System.err.println("发送失败[" + message + "]: " + exception.getMessage());
                    } else {
                        System.out.println("发送成功[" + message + "]"); //数据经过预处理，较为干净，注释降低print冗余
                    }
                });
//                Thread.sleep(50);  // 控制发送速率
            }
            csvReader.close();
        } catch (Exception e) {
            throw new RuntimeException("生产者异常", e);
        }
    }
}